package com.itheima.datastructure.blockingqueue;

import java.util.Arrays;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

/**
 * 双锁实现-阻塞队列
 *
 * @param <E> 元素类型
 */
@SuppressWarnings("all")
public class BlockingQueue2<E> implements BlockingQueue<E> {

    private final E[] array;
    private int head;
    private int tail;
    private AtomicInteger size = new AtomicInteger();

    //offer
    private ReentrantLock offerLock = new ReentrantLock();
    private Condition offerWaits = offerLock.newCondition();

    //poll
    private ReentrantLock pollLock = new ReentrantLock();
    private Condition pollWaits = pollLock.newCondition();

    public BlockingQueue2(int capacity) {
        this.array = (E[]) new Object[capacity];
    }

    private boolean isEmpty() {
        return size.get() == 0;
    }

    private boolean isFull() {
        return size.get() == array.length;
    }

    @Override
    public String toString() {
        return Arrays.toString(array);
    }


    @Override
    public void offer(E e) throws InterruptedException {
        int c; // 添加前元素个数
        offerLock.lockInterruptibly();
        try {
            // 1. 队列满则等待
            while (isFull()) {
                offerWaits.await();
            }

            // 2. 不满则入队   //tail有offerLock锁保护
            array[tail] = e;
            if (++tail == array.length) {
                tail = 0;
            }

            // 3. 修改 size ，如果添加一个元素后还有空位，则唤醒等待的 offer 线程
            c = size.getAndIncrement();
            if (c + 1 < array.length) {
                offerWaits.signal();
            }
        } finally {
            offerLock.unlock();
        }

        // 4. 队列 空-> 非空 则唤醒等待的poll线程
        if (c == 0) {
            //此处有多个offer线程，offer_1 offer_2 offer_3
            //只想要一个offer线程执行此操作
            pollLock.lock();
            try {
                pollWaits.signal();
            } finally {
                pollLock.unlock();
            }
        }
    }


    @Override
    public E poll() throws InterruptedException {
        E e;
        int c; // 取走前的元素个数
        pollLock.lockInterruptibly();
        try {
            // 1. 队列空则等待
            while (isEmpty()) {
                pollWaits.await();
            }

            // 2. 非空则出队  head有pollLock锁保护
            e = array[head];
            array[head] = null; // help GC
            if (++head == array.length) {
                head = 0;
            }

            // 3. 修改 size ,如果取走一个后还有元素可取，则唤醒等待的 poll 线程
            c = size.getAndDecrement();
            if (c > 1) {
                pollWaits.signal();
            }
        } finally {
            pollLock.unlock();
        }

        // 4. 队列 满->不满时  唤醒等待的offer线程
        if (c == array.length) {
            //此处有多个poll线程，poll_1 poll_2 poll_3
            //只想要一个poll线程执行此操作
            offerLock.lock();
            try {
                offerWaits.signal(); // ctrl+alt+t
            } finally {
                offerLock.unlock();
            }
        }

        return e;
    }


    @Override
    public boolean offer(E e, long timeout) throws InterruptedException {
        int c; // 添加前元素个数
        offerLock.lockInterruptibly();
        try {
            long nanos = TimeUnit.MILLISECONDS.toNanos(timeout);
            // 队列满则等待
            while (isFull()) {
                if (nanos <= 0) {
                    return false;
                }
                nanos = offerWaits.awaitNanos(nanos);
            }
            // 非满则入队
            array[tail] = e;
            if (++tail == array.length) {
                tail = 0;
            }
            c = size.getAndIncrement();
            // 如果添加一个元素后还有空位，则唤醒等待的 offer 线程
            if (c + 1 < array.length) {
                offerWaits.signal();
            }
        } finally {
            offerLock.unlock();
        }

        // 4. 队列 空-> 非空 则唤醒等待的poll线程
        if (c == 0) {
            //此处有多个offer线程，offer_1 offer_2 offer_3
            //只想要一个offer线程执行此操作
            pollLock.lock();
            try {
                pollWaits.signal();
            } finally {
                pollLock.unlock();
            }
        }
        return true;
    }


    public static void main(String[] args) throws InterruptedException {
        BlockingQueue2<String> queue = new BlockingQueue2<>(3);
        queue.offer("元素1");
        queue.offer("元素2");

        new Thread(() -> {
            try {
                queue.offer("元素3");
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        }, "offer").start();

        new Thread(() -> {
            try {
                queue.poll();
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        }, "poll").start();
    }
}
